package com.gitee.jmash.websocket.endpoint;


import com.gitee.jmash.websocket.model.Message;
import com.gitee.jmash.websocket.model.MsgType;
import com.gitee.jmash.websocket.model.SubscribeMessage;
import jakarta.websocket.Session;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 消息处理.
 *
 * @author cgd
 */
public class SubscribeHandle {

  private static Log log = LogFactory.getLog(SubscribeHandle.class);


  // 通道,SessionID
  private static Map<String, Set<String>> channels = new ConcurrentHashMap<>();

  // 通道,UserID,SessionID
  private static Map<String, Map<String, String>> userChannels = new ConcurrentHashMap<>();

  /**
   * 处理订阅和取消订阅.
   */
  public static synchronized void handle(Session session, Message message) {
    if (MsgType.subscribe.equals(message.getMsgType())) {
      subscribe(session, (SubscribeMessage) message);
    } else if (MsgType.unsubscribe.equals(message.getMsgType())) {
      unsubscribe(session, (SubscribeMessage) message);
    } else {
      log.info("NOT Handle:" + message);
    }
  }

  /**
   * Session Close.
   *
   * @param session
   */
  public static synchronized void close(Session session) {
    removeCloseSessionId(session);
  }

  /**
   * 订阅通道.
   */
  public static synchronized void subscribe(Session session, SubscribeMessage message) {
    if (message.getPrincipal() == null) {
      if (channels.containsKey(message.getChannelId())) {
        Set<String> sessionIds = channels.get(message.getChannelId());
        sessionIds.add(session.getId());
      } else {
        Set<String> sessionIds = new HashSet<>();
        sessionIds.add(session.getId());
        channels.put(message.getChannelId(), sessionIds);
      }
    } else {
      if (userChannels.containsKey(message.getChannelId())) {
        Map<String, String> userSessionMap = userChannels.get(message.getChannelId());
        userSessionMap.put(message.getPrincipal().getName(), session.getId());
      } else {
        Map<String, String> userSessionMap = new ConcurrentHashMap<>();
        userSessionMap.put(message.getPrincipal().getName(), session.getId());
        userChannels.put(message.getChannelId(), userSessionMap);
      }
    }
    Message resp = new Message();
    resp.setMsgType(MsgType.resp);
    resp.setChannelId(message.getChannelId());
    resp.setMessage("订阅成功！");
    SubscribeEndpoint.sendMessage(session.getId(), resp, false);
  }

  /**
   * 取消订阅通道.
   */
  public static synchronized void unsubscribe(Session session, SubscribeMessage message) {
    if (message.getPrincipal() == null) {
      if (channels.containsKey(message.getChannelId())) {
        Set<String> sessionIds = channels.get(message.getChannelId());
        sessionIds.remove(session.getId());
      }
    } else {
      if (userChannels.containsKey(message.getChannelId())) {
        Map<String, String> userSessionMap = userChannels.get(message.getChannelId());
        if (userSessionMap.containsKey(message.getPrincipal().getName())) {
          userSessionMap.remove(message.getPrincipal().getName());
        }
      }
    }
    Message resp = new Message();
    resp.setMsgType(MsgType.resp);
    resp.setChannelId(message.getChannelId());
    resp.setMessage("取消订阅成功！");
    SubscribeEndpoint.sendMessage(session.getId(), resp, false);
  }

  /**
   * 移除关闭的SessionId.
   */
  public static synchronized void removeCloseSessionId(Session session) {
    for (Entry<String, Set<String>> entry : channels.entrySet()) {
      Iterator<String> iterator = entry.getValue().iterator();
      while (iterator.hasNext()) {
        String sessionId = iterator.next();
        if (sessionId.equals(session.getId())) {
          iterator.remove();
        }
      }
    }
    for (Entry<String, Map<String, String>> channelEntry : userChannels.entrySet()) {
      Map<String, String> userSessionMap = channelEntry.getValue();
      Iterator<Entry<String, String>> iterator = userSessionMap.entrySet().iterator();
      while (iterator.hasNext()) {
        Entry<String, String> userEntry = iterator.next();
        if (userEntry.getValue().equals(session.getId())) {
          iterator.remove();
        }
      }
    }
  }

  /**
   * 获取该通道全部SessionIds.
   *
   * @param channelId 通道ID
   */
  public static Set<String> getChannelSession(String channelId) {
    Set<String> sessionIds = new HashSet<>();
    if (channels.containsKey(channelId)) {
      sessionIds.addAll(channels.get(channelId));
    }
    if (userChannels.containsKey(channelId)) {
      Map<String, String> userSessionMap = userChannels.get(channelId);
      sessionIds.addAll(userSessionMap.values());
    }
    return sessionIds;
  }

  /**
   * 获取该通道某个用户SessionId.
   *
   * @param channelId 通道ID
   * @param userId    用户ID.
   */
  public static String getChannelUserSession(String channelId, String userId) {
    if (userChannels.containsKey(channelId)) {
      Map<String, String> userSessionMap = userChannels.get(channelId);
      if (userSessionMap.containsKey(userId)) {
        return userSessionMap.get(userId);
      }
    }
    return null;
  }


  /**
   * 给指定通道发送消息.
   */
  public static int sendMessage(Message message, boolean sync) {
    Set<String> sessionIds = getChannelSession(message.getChannelId());
    if (sessionIds.isEmpty()) {
      return 0;
    }
    return SubscribeEndpoint.sendMessages(sessionIds, message, sync);
  }

  /**
   * 给指定通道的用户发送消息.
   */
  public static boolean sendMessage(String userId, Message message, boolean sync) {
    String sessionId = getChannelUserSession(message.getChannelId(), userId);
    if (sessionId == null) {
      return false;
    }
    return SubscribeEndpoint.sendMessage(sessionId, message, sync);
  }

}
